@djaglowski Do you have an idea how to do it technically? Not really sure how the test scenario should look like.

Do you want me to create a receiver which consists of several operators, and then validate return value of the last receiver?

You can build and start a pipeline at the pkg/stanza level that is basically just a slice of operators. (Default operator not required)

This PR was marked stale due to lack of activity. It will be closed in 14 days.

Foresight Summary

    
Major Impacts

build-and-test-windows duration(4 seconds) has decreased 45 minutes 13 seconds compared to main branch avg(45 minutes 17 seconds).

🔎 See details on Foresight

^{*You can configure Foresight comments in your organization settings page.}

@djaglowski Is this change still wanted according to the #17079?

My understanding is that the issue is resolved by #17079.

The ConsumeLogs function is now only blocked when waiting to write to a channel, whereas we formerly were blocking with an expectation of being able to read from a channel. I would certainly appreciate your take on this as well, if you have time. Either way, thanks for all the effort you put into this PR.

@djaglowski I rebased the PR. Backpropagation is added, but it's not used in logstransformprocessor for now.

Looking at the #17079 change, I'm not sure if asynchronous processors are align with the pipeline architecture. In current state it will take all logs and claim that they are processed correctly:

I can revert this to the desired state of my commit, but not sure what is the decision here.

I'm not sure if asynchronous processors are align with the pipeline architecture.

I may be wrong, but I don't think processors are necessarily expected to be synchronous. For example, batchprocessor is one of only two in the core collector and it is asynchronous.

In current state it will take all logs and claim that they are processed correctly

Can you elaborate on this? I'm not familiar with how this is communicated and what impact it has.

In current state it will take all logs and claim that they are processed correctly

Can you elaborate on this? I'm not familiar with how this is communicated and what impact it has.

I'm not sure if asynchronous processors are align with the pipeline architecture.

I may be wrong, but I don't think processors are necessarily expected to be synchronous. For example, batchprocessor is one of only two in the core collector and it is asynchronous.

There is an error backpropagation:

https://github.com/open-telemetry/opentelemetry-collector/blob/e4e76dec22ecf8cf1d5b70af6d74215f0e05f0c0/processor/processorhelper/logs.go#L65-L71

Not really sure who and how handle the errors, but reminds me there is some issue with batchprocessor do not backpropagating the errors:

When exporter cannot handle more data and has to drop them it returns error to batchprocessor, but it is not backpropagated to receiver, so we cannot handle it (stop receiving data, ask client to stop sending them for some time). (@swiatekm-sumo correct me if I'm wrong)

Anyway, I think the PR is ready and making logstransformprocessor sync or not should be matter of another PR.

In order to justify the additional complexity, we need to have a clear design for how this would be used. It's no longer clear to me what these changes would accomplish. What will we do with the returned integer, specifically?

When exporter cannot handle more data and has to drop them it returns error to batchprocessor, but it is not backpropagated to receiver, so we cannot handle it (stop receiving data, ask client to stop sending them for some time). (@swiatekm-sumo correct me if I'm wrong)

Yeah, this is correct, and tracked in open-telemetry/opentelemetry-collector#4646.

In general, if we have an asynchronous component, that component essentially acts as a buffer and needs to take care of its own retry logic. Being asynchronous, it can't return errors to the caller, but it can and should retry on errors it gets from the rest of the pipeline. Otherwise it just drops data without providing any feedback and stops conducting backpressure.

This PR was marked stale due to lack of activity. It will be closed in 14 days.

@djaglowski any input here? I don't feel this change makes sense in other scenarios (maybe some metric to point how many data has been dropped in the queue)

In general, if we have an asynchronous component, that component essentially acts as a buffer and needs to take care of its own retry logic. Being asynchronous, it can't return errors to the caller, but it can and should retry on errors it gets from the rest of the pipeline. Otherwise it just drops data without providing any feedback and stops conducting backpressure.

Thanks for explaining this, it makes sense to me now why this processor would need special considerations for handling backpressure.

Do we have the same problem in the pkg/stanza receivers? They all share the same conversion/batching strategy. This design has been questioned in the past, mostly for its complexity. I wonder if the correct solution here is to make the shared converter synchronous.

Do we have the same problem in the pkg/stanza receivers? They all share the same conversion/batching strategy. This design has been questioned in the past, mostly for its complexity. I wonder if the correct solution here is to make the shared converter synchronous.

It seems that errors are only logged out by the receiver. There is no try to handle recoverable errors or to inform client that data has been dropped

https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/pkg/stanza/adapter/receiver.go#L134-L139

This PR was marked stale due to lack of activity. It will be closed in 14 days.

It seems that errors are only logged out by the receiver. There is no try to handle recoverable errors or to inform client that data has been dropped

@djaglowski Should I create separate issue for this?

@sumo-drosiek, I think a separate issue makes sense.

This PR was marked stale due to lack of activity. It will be closed in 14 days.

Closed as inactive. Feel free to reopen if this PR is still being worked on.